package com.globefrog.framework.juc.executor;

import com.globefrog.framework.common.exception.CustomExecutorException;
import com.globefrog.framework.juc.proxy.CallableProxy;
import com.google.common.collect.Lists;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;


/**
 * 自定义线程.
 *
 * @author : caijinbang
 * @date : Create in 2021-01-30
 */
public class CustomExecutor extends BaseExecutor {

  /**
   * 构造函数.
   *
   * @param executorService 线程池
   */
  public CustomExecutor(ExecutorService executorService) {
    super(executorService);
  }

  /**
   * 定长线程池.
   *
   * @param threadPoolSize 线程池长度
   * @return 自定义线程池
   */
  public static CustomExecutor fixedSizeExecutor(int threadPoolSize) {
    return new CustomExecutor(Executors.newFixedThreadPool(threadPoolSize));
  }

  /**
   * 可缓存线程池.
   *
   * @return 可缓存线程池
   */
  public static CustomExecutor cachedExecutor() {
    return new CustomExecutor(Executors.newCachedThreadPool());
  }

  /**
   * callable加入线程池中.
   *
   * @param callable callable
   * @param <T>      结果对象
   * @return 结果对象
   */
  public <T> Future<T> execute(Callable<T> callable) {
    if (isRunning()) {
      // 实例化callable
      Callable<T> callableBean = initBean(callable);
      // 将callable包裹在代理类中,对线程池日志进行初始化
      CallableProxy<T> proxy = new CallableProxy<T>(callableBean, logHelper, errorHandler, false);
      return executorService.submit(proxy);
    } else {
      throw new RejectedExecutionException();
    }
  }


  /**
   * callable加入线程池中.日志进行分离
   *
   * @param callable callable
   * @param <T>      结果对象
   * @return 结果对象
   */
  public <T> Future<T> executeSeparately(Callable<T> callable) {
    if (isRunning()) {
      // 实例化callable
      Callable<T> callableBean = initBean(callable);
      // 将callable包裹在代理类中,对线程池日志进行初始化
      CallableProxy<T> proxy = new CallableProxy<T>(callableBean, logHelper, errorHandler, true);
      return executorService.submit(proxy);
    } else {
      throw new RejectedExecutionException();
    }
  }

  /**
   * callable加入线程池中.
   *
   * @param callable callable
   * @param logAble  是否打印日志
   * @param <T>      结果对象
   * @return 结果对象
   */
  public <T> Future<T> executeSeparately(Callable<T> callable, boolean logAble) {
    if (isRunning()) {
      Callable<T> callableBean = initBean(callable);
      CallableProxy<T> proxy = new CallableProxy<>(callableBean, logHelper, errorHandler, true);
      return executorService.submit(logAble ? proxy : callableBean);
    } else {
      throw new RejectedExecutionException();
    }
  }

  /**
   * 多个call执行
   *
   * @param callables callables
   * @param <T>       结果对象
   * @return 结果对象
   */
  public <T> List<T> executeAllSeparately(List<? extends Callable<T>> callables) {
    if (isRunning()) {
      return doExecute(callables, true, true);
    } else {
      throw new RejectedExecutionException();
    }
  }

  /**
   * 多个call执行
   *
   * @param callables callables
   * @param <T>       结果对象
   * @return 结果对象
   */
  public <T> List<T> executeAllSeparately(List<? extends Callable<T>> callables, boolean logAble) {
    if (isRunning()) {
      return doExecute(callables, true, logAble);
    } else {
      throw new RejectedExecutionException();
    }
  }

  public <T> List<T> executeAll(List<? extends Callable<T>> callables) {
    if (isRunning()) {
      return doExecute(callables, false, true);
    } else {
      throw new RejectedExecutionException();
    }
  }

  protected <T> List<T> doExecute(List<? extends Callable<T>> callables, boolean logSeparate,
      boolean logAble) {
    if (isRunning()) {
      try {
        List<Callable<T>> proxyTasks = createProxyTasks(callables, logSeparate, logAble);
        List<Future<T>> futures = executorService.invokeAll(proxyTasks);
        List<T> results = new ArrayList(futures.size());

        for (Future<T> future : futures) {
          if (isRunning()) {
            results.add(future.get());
          } else if (future.isDone()) {
            results.add(future.get());
          } else {
            future.cancel(true);
          }
        }
        return results;
      } catch (ExecutionException | InterruptedException e) {
        throw new CustomExecutorException(e);
      }
    } else {
      throw new RejectedExecutionException();
    }
  }

  private <T> List<Callable<T>> createProxyTasks(List<? extends Callable<T>> callables,
      boolean logSeparate, boolean logAble) {
    if (isRunning()) {
      List<Callable<T>> proxys = Lists.newArrayListWithCapacity(callables.size());
      for (Callable<T> callable : callables) {
        Callable<T> taskBean = initBean(callable);
        if (logAble) {
          proxys.add(new CallableProxy<>(taskBean, logHelper, errorHandler, logSeparate));
        } else {
          proxys.add(taskBean);
        }
      }
      return proxys;
    } else {
      throw new RejectedExecutionException();
    }
  }
}
